package com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.utils

import org.apache.spark.sql.{DataFrame, SparkSession}

class brandCase {
  def brandCaseWhen(data:DataFrame,spark:SparkSession): DataFrame ={

    data.registerTempTable("brandCaseData")
    spark.sql(
      """
        |select *,
        |case
        |when firstCategoryId = '10019' and brandName_cn = "其它" or brandName_cn = "其他" or brandName_cn = '-1' then "其他/OTHER"
        |when firstCategoryId = '10019' and brandName_cn = "Kiehl＇s/科颜氏" or brandName_cn = "科颜氏/KIEHL S"  then "科颜氏/KIEHLS"
        |when firstCategoryId = '10019' and brandName_cn = "SU:M37°/苏秘37°" or brandName_cn = "SU:M37˚苏秘37˚/SU:M37˚37˚"  then "苏秘37/SU:M37°"
        |when firstCategoryId = '10019' and brandName_cn = "苏秘37°" or brandName_cn = "苏秘"  then "苏秘37/SU:M37°"
        |when firstCategoryId = '10019' and brandName_cn = "苏秘37°SU:M37°/37°SU:M37°" or brandName_cn = "苏秘37°/SU:M37°"  then "苏秘37/SU:M37°"
        |
        |else brandName_cn end cn
        |from brandCaseData
      """.stripMargin)
      .registerTempTable("brandCN")


    val result = spark.sql(
      s"""
         |select *,
         |case when split(brandName_cn,"\\/")[1] is not null or brandName_cn rlike '\\/'
         |then concat(upper(split(brandName_cn,"\\/")[1]),"/",upper(split(brandName_cn,"\\/")[0]))
         |else upper(brandName_cn) end en
         |from brandCN
        """.stripMargin)
      .drop("brandName_en")
      .withColumnRenamed("en", "brandName_en")
    return result
  }

}
